package com.zp.datax.core;

import com.zp.datax.core.function.ConsumerWrapper;
import com.zp.datax.core.util.ThreadPoolFactory;
import lombok.*;
import lombok.extern.slf4j.Slf4j;

import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 自主研发时间轮调度器
 *
 * @author bruce
 * @date 2019-07-15
 */
@Slf4j
public class TimingWheel {

    /**
     * 默认值：固定60为一个周期
     */
    private final static int DEFAULT_CYCLE = 60;
    /**
     * 默认值：map初始容量
     */
    private final static int DEFAULT_INITIAL_CAPACITY = 256;
    /**
     * 存储任务信息，key为触发时刻，value为任务具体触发信息
     */
    private volatile Map<Integer, Item> table;
    /**
     * 存储对应任务执行周期，key剩余时间（wheel上的时刻），value对应同一周期不同的任务链
     */
    private volatile Map<Integer, Set<Integer>> wheel;
    /**
     * 存储上次已经执行过的任务，用于清除标记
     */
    private BlockingQueue<ExpireTask> expire_queue = new DelayQueue<>();
    /**
     * 当前轮上的指针指向
     */
    private AtomicInteger pointer = new AtomicInteger(1);
    /**
     * 任务线程暂停标识
     */
    private volatile boolean timingStop = false;
    /**
     * 清理线程暂停标识
     */
    private volatile boolean cleanStop = false;
    /**
     * 管理执行任务
     */
    private BlockingQueue<Integer> trigger_queue = new LinkedBlockingQueue<>(256);


    /**
     * 构造函数
     *
     * @param map 初始化Map
     */
    public TimingWheel(Map<Integer, String> map) {
        this(map.size());
    }


    /**
     * 默认构造函数
     */
    public TimingWheel() {
        this(DEFAULT_INITIAL_CAPACITY);
    }

    /**
     * 构造函数
     *
     * @param initialCapacity 初始容量值
     */
    public TimingWheel(int initialCapacity) {
        this.table = new ConcurrentHashMap<>(initialCapacity);
        createWheel();
    }

    public BlockingQueue<Integer> getTriggerQueue() {
        return trigger_queue;
    }

    /**
     * 创建无任务信息的固定大小周期轮
     */
    private void createWheel() {
        wheel = new ConcurrentHashMap<>(DEFAULT_CYCLE);

        for (int i = 1; i <= DEFAULT_CYCLE; i++) {
            wheel.put(i, new HashSet<>());
        }
    }

    /**
     * 删除job任务
     *
     * @param taskId 任务编号
     */
    public Item removeTriggerTask(final Integer taskId) {
        return table.remove(taskId);
    }

    /**
     * 开启任务线程和清理线程
     * 任务线程：检查任务触发时机并加入触发执行对列
     * 清理线程：将已经触发执行任务移除触发对列
     */
    public void start() {

        ThreadPoolExecutor pool = ThreadPoolFactory.INSTANCE.getThreadPool();

        pool.execute(() -> {
            while (!timingStop) {
                try {

                    TimeUnit.MILLISECONDS.sleep(1000L);
                    checkTriggerTask(pointer.get());

                } catch (Exception e) {
                    if (!timingStop) {
                        log.error("执行调度中 :: +++++++++++++++ : {}", e.getMessage());

                    }
                }

                incrementSetPointer();
            }
        });


        pool.execute(() -> {
            while (!cleanStop) {
                try {

                    //todo 目前单个移除过期任务，如果已经处理的过期任务较多，可采用批量方式移除
                    ExpireTask task = expire_queue.take();
                    wheel.get(Integer.valueOf(task.getPointer())).remove(Integer.valueOf(task.getTaskId()));

                } catch (Exception e) {
                    if (!cleanStop) {
                        log.error("执行过期任务清理中 :: +++++++++++++++ : {}", e.getMessage());
                    }
                }
            }
        });

    }

    /**
     * 自增计数指针指向
     */
    private void incrementSetPointer() {
        //60为一周期，计数到达60从1开始
        pointer.incrementAndGet();
        if (pointer.compareAndSet(DEFAULT_CYCLE + 1, 1)) {
            return;
        }
    }

    /**
     * 具体定时触发逻辑从此开始，根据当前指针指向检查当前任务是否到达触发时机
     *
     * @param pointer 指针指向
     */
    private void checkTriggerTask(final int pointer) {

        Set<Integer> taskIds = wheel.get(Integer.valueOf(pointer));
        if (taskIds.isEmpty()) {
            return;
        }

        ConsumerWrapper<Integer, Exception> consumerWrapper = this::setTable;

        taskIds.forEach(consumerWrapper.andThen(taskId -> {
        }));

    }

    /**
     * 修改计划
     *
     * @param newItem
     * @param oldItem
     */
    private void changeWheel(Item oldItem, Item newItem) {

        log.info("oldItem :: {}", oldItem);
        log.info("newItem :: {}", newItem);

        Integer taskId = newItem.getTaskId();

        //初次添加调度计划
        if (Objects.isNull(oldItem)
                && wheel.get(Integer.valueOf(String.valueOf(newItem.getSecond()))).add(taskId)) {
            return;
        }


        Integer oldKey = Integer.valueOf(String.valueOf(oldItem.getSecond()));
        Set<Integer> set = wheel.get(oldKey);

        //新旧调度计划一致时，并且taskId已经存在调度池中，则调度计划不变
        if (newItem.equals(oldItem)
                && set.contains(taskId)) {
            return;
        }

        //新旧调度计划不一致时，将新调度计划添加调度池，当返回false-说明旧调度计划已经存在并且与新调度计划调度时刻一致，则返回不作出改变
        //当返回true-新调度计划添加成功，后续继续执行删除旧调度计划
        if (!wheel.get(Integer.valueOf(String.valueOf(newItem.getSecond()))).add(taskId)) {
            return;
        }

        //删除旧的调度计划
        if (!set.isEmpty()) {
            expire_queue.add(new ExpireTask(oldKey, taskId, 1000L));
        }

    }

    /**
     * 设置周期
     *
     * @param taskId
     */
    private void setTable(final Integer taskId) {
        Item oldValue = table.get(taskId);

        //为空删除
        if (Objects.isNull(oldValue)) {
            log.warn("+++++++++++++++ 将 taskId :: {} 移除计划", taskId);
            expire_queue.add(new ExpireTask(pointer.get(), taskId, 1000L));
            return;
        }


        long round = oldValue.getRound();
        int second = (int) oldValue.getSecond();

        if (round == 0L && second == pointer.get()) {
            trigger_queue.add(taskId);
        } else {
            round--;
            oldValue.setRound(round);
        }
    }

    /**
     * 推送任务
     *
     * @param taskId 任务编号
     * @param remainingTime 距离下次执行还剩余多少秒
     */
    public void pushTriggerTask(final Integer taskId, final long remainingTime) {

        Item oldValue = table.get(taskId);
        Item newValue = createItem(taskId, remainingTime);

        if (Objects.isNull(oldValue)) {
            table.put(taskId, newValue);
        } else {
            table.replace(taskId, oldValue, newValue);
        }

        changeWheel(oldValue, newValue);

    }

    /**
     * 计算触发周期时间点
     *
     * @param remainingTime 距离触发剩余时间，单位秒
     * @return
     */
    private Item createItem(final Integer taskId, final long remainingTime) {
        long mod = remainingTime % DEFAULT_CYCLE;
        long round = remainingTime / DEFAULT_CYCLE;

        int sum = (int) mod + pointer.get();
        int over = sum - DEFAULT_CYCLE;
        int second = over > 0 ? over : over < 0 ? sum : DEFAULT_CYCLE;

        if (mod == 0) {
            round--;
        }

        return new Item(taskId, (round < 0L ? 0L : round), second);
    }

    /**
     * destroy
     */
    public void destroy() {
        if (!timingStop) {
            timingStop = true;
        }
        if (!cleanStop) {
            cleanStop = true;
        }
    }

    @Data
    @AllArgsConstructor(access = AccessLevel.PACKAGE)
    final static class Item {

        //任务编号
        final int taskId;
        //属性最多多少轮，单位60秒为一轮
        long round;
        //属性秒，单位1秒
        long second;

    }

    @Getter
    @RequiredArgsConstructor(access = AccessLevel.PACKAGE)
    final static class ExpireTask implements Delayed {

        //指针指向
        final int pointer;
        //任务编号
        final int taskId;
        //执行时间
        long executeTime;

        ExpireTask(int pointer, int taskId, long expireTime) {
            this(pointer, taskId);
            this.executeTime = TimeUnit.NANOSECONDS.convert(expireTime, TimeUnit.MILLISECONDS) + System.nanoTime();
        }

        @Override
        public final long getDelay(TimeUnit unit) {
            return unit.convert(executeTime - System.nanoTime(), TimeUnit.NANOSECONDS);
        }

        @Override
        public final int compareTo(Delayed o) {
            ExpireTask task = (ExpireTask) o;

            return Integer.valueOf(pointer).compareTo(Integer.valueOf(task.getPointer()));
        }

    }

}
